use prost::{encode_length_delimiter, length_delimiter_len};
use bytes::{BufMut, BytesMut};

#[derive(PartialEq, Clone, Copy, Debug)]
pub enum LogRecordType {
    // PUT 操作
    NORMAL = 1,
    // 删除被标记数据
    DELETED = 2,
}


/// LogRecord 写入到数据文件的记录
/// 之所以叫日志，是因为数据文件中的数据是追加写入的，类似日志的格式
pub struct LogRecord {
    pub(crate) key: Vec<u8>,
    pub(crate) value: Vec<u8>,
    pub(crate) rec_type: LogRecordType,
}

/// 数据位置索引信息，描述数据存储到了哪个位置
#[derive(Clone, Copy, Debug)]
pub struct LogRecordPos { 
    pub(crate) offset: u64, // 偏移量
    pub(crate) file_id: u32, // 文件编号
}


/// 从数据文件中读取的 log_record 信息，包含其 size
pub struct ReadLogRecord {
    pub(crate) record: LogRecord,
    pub(crate) size: usize,
}

impl LogRecord {

    /// 编码 LogRecord，返回字节数组，编码后的格式如下
    ///
    ///	+-------------+-------------+--------------+-------------+--------------+--------------+
    ///	|  type 类型   |    key size |   value size |      key    |      value   |  crc 校验值   |
    ///	+-------------+-------------+--------------+-------------+--------------+--------------+
    ///     1字节        变长（最大5）   变长（最大5）        变长            变长           4字节
    pub fn encode(&self) -> Vec<u8> {
        let (enc_buf, _) = self.encode_and_get_crc();
        enc_buf
    }

    pub fn get_crc(&self) -> u32 {
        let (_, crc) = self.encode_and_get_crc();
        crc
    }

    fn encode_and_get_crc(&self) -> (Vec<u8>, u32) {

        let mut buf = BytesMut::new();

        buf.reserve(self.encoded_length());
        // type位置
        buf.put_u8(self.rec_type as u8);

        // key size
        encode_length_delimiter(self.key.len(), &mut buf).unwrap();
        // value size
        encode_length_delimiter(self.value.len(), &mut buf).unwrap();
        // key
        buf.extend_from_slice(&self.key);
        // value
        buf.extend_from_slice(&self.value);

        // crc 校验
        let mut hasher = crc32fast::Hasher::new();
        hasher.update(&buf);
        let crc = hasher.finalize();
        buf.put_u32(crc);

        (buf.to_vec(), crc)
    }

    fn encoded_length(&self) -> usize {
        std::mem::size_of::<u8>() +
        length_delimiter_len(self.key.len()) +
        length_delimiter_len(self.value.len()) +
        self.key.len() +
        self.value.len() +
        4
    }

}

impl LogRecordType {
    pub fn from_u8(v: u8) -> Self {
        match v {
            1 => LogRecordType::NORMAL,
            2 => LogRecordType::DELETED,
            _ => {
                println!("type = {:?}", v);
                panic!("unknown log record type");
            }
        }
    }
}

impl From<u8> for LogRecordType {
    fn from(value: u8) -> Self {
        match value {
            1 => LogRecordType::NORMAL,
            2 => LogRecordType::DELETED,
            _ => {
                println!("type = {:?}", value);
                panic!("unknown log record type");
            }
        }
    }
}


/// 获取 LogRecord header 部分的最大长度
pub fn max_log_record_header_size() -> usize {
    std::mem::size_of::<u8>() + length_delimiter_len(std::u32::MAX as usize) * 2
}



#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_encode_log_record() {
        // 1.正常情况
        let rec1 = LogRecord {
            key: "name".as_bytes().to_vec(),
            value: "bitcask-rs".as_bytes().to_vec(),
            rec_type: LogRecordType::NORMAL,
        };
        let enc1 = rec1.encode();
        assert!(enc1.len() > 0);
        assert_eq!(1020360578, rec1.get_crc());

        // 2.value 为空的情况
        let rec2 = LogRecord {
            key: "name".as_bytes().to_vec(),
            value: Default::default(),
            rec_type: LogRecordType::NORMAL,
        };
        let enc2 = rec2.encode();
        assert!(enc2.len() > 0);
        assert_eq!(3756865478, rec2.get_crc());

        // 3.类型为 deleted 的情况
        let rec3 = LogRecord {
            key: "name".as_bytes().to_vec(),
            value: Default::default(),
            rec_type: LogRecordType::DELETED,
        };
        let enc3 = rec3.encode();
        assert!(enc3.len() > 0);
        assert_eq!(3993316699, rec3.get_crc());
    }

    #[test]
    fn test_u8_into_type () {

        let type1 = 1_u8;
        assert_eq!(LogRecordType::NORMAL, type1.into());
    }
}

